#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/socket.h>

#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <netdb.h>
#include <pthread.h> 
#include <time.h>
#include <signal.h>
#include <string.h>
#include <math.h>
#include <linux/unistd.h>
#include <sys/syscall.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sched.h>
#include <signal.h>
#include <setjmp.h>
#include <errno.h>
#include <assert.h>


#include "SDL/SDL.h"
#include "SDL/SDL_image.h"
#include "SDL/SDL_ttf.h"


#define LAM_PORT 36668
#define HEARTBEATS_PORT 36667
#define SERVER_PORT 36666
#define BUFLEN 512
#define GATE_NUM 28
#define T_MAX 20 
#define WIDTH 1024
#define HEIGHT 768
#define TT_SIGUSR1 (SIGUSR1)
#define TT_SIGUSR2 (SIGUSR2)

#define REQ 1
#define OK 2
#define TIME_INTERVAL_1 1
#define TIME_INTERVAL_2 2
#define TIME_INTERVAL_10 10
#define STATE_NO_CARE 1
#define STATE_REQ 2
#define STATE_CR 3
#define STATE_DONE 4

#define MAX_THREAD_NUM  20
#define ACTION_INCREMENT 1
#define ACTION_SET 2
#define ACTION_NO_ADD 3

int lamport_action = 0;
int count = 0;
int lamport_agree_count = 0;
int lamport_queue_agree_count = 0;
int lamp_counter = 0;
pthread_mutex_t lamp_mut; 
pthread_mutex_t lamp_seg_mut; 

long g_my_sec = 0;
long g_my_usec = 0;
pid_t g_my_pid = 0;
int g_seg_counter = 0;


char arg0[64];
char arg1[8]= "\0";
char arg2[8]= "\0";
char arg3[8]= "\0";
char pname[BUFLEN];

int g_my_port = 0;
int semaph = 0;
int seg_num = 0;
int total_seg_num = 0;
int target = -1;
int lamport_count = 0;
int lamport_switch = 0;
char g_my_ip[BUFLEN] = "\0";
char g_other_ip[BUFLEN];
int g_other_port = 0;
typedef struct
{
	int state;
	char desc[10];
} LAMPORT_STATE;
LAMPORT_STATE lam_state;

struct seg_arguments
{
  unsigned int minval;	//min number of segments
  unsigned int maxval;	//max number of segments
  unsigned int thbeat;	//heartbeat time
}arg;

typedef struct node *link_seg;
typedef struct 
{
	char ip[BUFLEN];
	int  port;
	long last_time;
	long sec;
	long u_sec;
} seginfo;

struct node {
    seginfo seg;
    link_seg next;
};
typedef struct lamp_node *lamp_link_seg;

typedef struct 
{
	char ip[BUFLEN];
	int  pid;
} lamp_seginfo;


struct lamp_node {
    lamp_seginfo lamp_seg;
    lamp_link_seg next;
};


static link_seg head =NULL;
static lamp_link_seg lamp_head =NULL;

pthread_mutex_t mut; 

struct in_addr localInterface;
//struct sockaddr_in kill_recv_group;
struct sockaddr_in kill_last_groupSock;
struct sockaddr_in kill_all_groupSock;
struct sockaddr_in heartbeat_group_sock, s_addr_in;
struct sockaddr_in lamport_group_sock;
int lamp_socket;

pid_t srv_listen = -1;
int maxtime = 0;
int g_num_process = 0;
int g_kill_me = 0;
//pthread_mutex_t mut; 
char ip[BUFLEN];

pthread_t tid[T_MAX];
pthread_t g_thread;
pthread_t g_heartbeat_thread;
pthread_t g_display_thread;
pthread_t g_kill_recv_thread;
pthread_t g_interval_check_thread;


SDL_Surface *background = NULL;
SDL_Surface *criticalregion1 = NULL;
SDL_Surface *criticalregion2 = NULL;
SDL_Surface *criticalregion3 = NULL;
SDL_Surface *message = NULL;

TTF_Font *font = NULL;
SDL_Color textColor = { 255, 255, 255 };
SDL_Color textColor1 = { 255, 0, 0 };
pthread_t g_lamport_thread;
FILE *g_test;

//global variables for all the child processes
pid_t heartbeat_pid = -1 ;
pid_t seglogic_pid = -1 ;
pid_t display_pid = -1 ;
	
char kill_msg;
int msg_type;
int lamport_cr;
int lamport_display_switch = 0;
struct sockaddr_in group_sock;
char buf[BUFLEN]="0";
int socket_heartbeat=0;
FILE *fp = NULL;  //the log file
int seg_cnt = 0;
char seg[1024]="\0";


char gates_adr[GATE_NUM][20] = {	//to be replaced with real lab nodes addresses
	"tile-0-0.local", 
	"tile-0-1.local", 
	"tile-0-2.local", 
	"tile-0-3.local",
	"tile-1-0.local", 
	"tile-1-1.local", 
	"tile-1-2.local",
	"tile-1-3.local", 
	"tile-2-0.local", 
	"tile-2-1.local", 
	"tile-2-2.local",
	"tile-2-3.local", 
	"tile-3-0.local", 
	"tile-3-1.local",
	"tile-3-2.local",
	"tile-3-3.local", 
	"tile-4-0.local", 
	"tile-4-1.local", 
	"tile-4-2.local",
	"tile-4-3.local", 
	"tile-5-0.local", 
	"tile-5-1.local",
	"tile-5-2.local", 
	"tile-5-3.local",
	"tile-6-0.local", 
	"tile-6-1.local", 
	"tile-6-2.local", 
	"tile-6-3.local"	};

link_seg make_node (seginfo seg)
{
	link_seg p = malloc(sizeof(struct node));
	strcpy(p->seg.ip, seg.ip);
	p->seg.port = seg.port;
	p->seg.last_time= seg.last_time;
	p->seg.sec = seg.sec;
	p->seg.u_sec = seg.u_sec;
	p->next = NULL;
	return p;
}
lamp_link_seg lamp_make_node (lamp_seginfo lamp_seg)
{
	lamp_link_seg p = malloc(sizeof(struct node));
	strcpy(p->lamp_seg.ip, lamp_seg.ip);
	p->lamp_seg.pid = lamp_seg.pid;

	p->next = NULL;
	return p;
}

void append_node(link_seg head, link_seg p)
{
	if(NULL == head)
	{
		p->next = NULL;
	}
	else
	{
		p->next = head->next;
	}
	head->next = p;
	
}
void free_node(link_seg p)
{
	free(p);
}
void lamp_free_node(lamp_link_seg p)
{
	free(p);
}

link_seg search(seginfo seg)
{
	link_seg p;
	for(p=head;p;p=p->next)
	{
		if((strcmp(seg.ip, p->seg.ip) == 0) && (p->seg.port == seg.port))
		{
			p->seg.last_time= seg.last_time;
			return p;
		}
	}
	return NULL;
}
lamp_link_seg lamp_search(lamp_seginfo lamp_seg)
{
	lamp_link_seg p;
	for(p=lamp_head;p;p=p->next)
	{
		if((strcmp(lamp_seg.ip, p->lamp_seg.ip) == 0) && (p->lamp_seg.pid == lamp_seg.pid))
		{

			return p;
		}
	}
	return NULL;
}

void insert(link_seg p)
{
	p->next = head;
	head = p;
}
void lamp_insert(lamp_link_seg p)
{
	p->next = lamp_head;
	lamp_head = p;
}

void lamp_delete_from_list(lamp_link_seg p)
{
	lamp_link_seg pre;
	if(p == lamp_head)
	{
		lamp_head = p->next;
		return;
	}
	for(pre=lamp_head; pre; pre = pre->next)
	{
		if(pre->next == p )
		{
			pre->next = p->next;
			return;
		}
	}
}

void delete_from_list(link_seg p)
{
	link_seg pre;
	if(p == head)
	{
		head = p->next;
		return;
	}
	for(pre=head; pre; pre = pre->next)
	{
		if(pre->next == p )
		{
			pre->next = p->next;
			return;
		}
	}
}

void traverse(void (*visit)(link_seg))
{
	link_seg p;
		
	for(p=head;p;p=p->next)
	{
		visit(p);
	}
	return;
}
void destroy(void)
{
	link_seg q,p=head;
	while(p)
	{
		q=p;
		p=p->next;
		free_node(q);
	}
	return;
}
void push(link_seg p)
{
	insert(p);
}

link_seg pop(void)
{
	if(NULL == head)
	{
		return NULL;
	}
	else
	{
		link_seg p=head;
		head = head->next;
		return p;
	}
}

int loginfo(char *pstr)
{
	time_t lt;
	char *ptm ;
	char buf[1024];
	int i, cnt= 0;
	struct tm *p;
	lt = time(NULL);
	p = localtime(&lt);
	ptm = ctime(&lt);
	
	char hostname[1024]="";

	gethostname(hostname, 1024);

	sprintf(buf, "[%s]%d-%d-%d %d:%d:%d %s\n", hostname, 1900+p->tm_year, 1 + p->tm_mon, p->tm_mday,  p->tm_hour, p->tm_min, p->tm_sec, pstr);
	
	for ( i = 0; buf[i] != '\0'; i++ ) 
	{
		cnt++;
	}
	fwrite(buf,cnt, 1, fp);	
	i = fflush(fp);
	return 0;
}


timer_t create_timer(int signo) 
{
    timer_t timerid;
    struct sigevent se;
    se.sigev_notify=SIGEV_SIGNAL;
    se.sigev_signo = signo;
    if (timer_create(CLOCK_REALTIME, &se, &timerid) == -1) 
    {
        perror("Failed to create timer");
        exit(-1);
    }
    return timerid;
}

void set_timer(timer_t timerid, int seconds) 
{
    struct itimerspec timervals;
    timervals.it_value.tv_sec = seconds;
    timervals.it_value.tv_nsec = 0;
    timervals.it_interval.tv_sec = seconds;
    timervals.it_interval.tv_nsec = 0;

    if (timer_settime(timerid, 0, &timervals, NULL) == -1) 
    {
        loginfo("Failed to start timer");
        exit(1);
    }
    return;
}

void install_sighandler(int signo, void(*handler)(int)) 
{
    sigset_t set;
    struct sigaction act;

    /* Setup the handler */
    act.sa_handler = handler;
    act.sa_flags = SA_RESTART;
    sigaction(signo, &act, 0);

    /* Unblock the signal */
    sigemptyset(&set);
    sigaddset(&set, signo);
    sigprocmask(SIG_UNBLOCK, &set, NULL);

    return;
}





int get_app_name(char *p, char *pname)
{
	char *q;
	int len = 0, lenp = 0;
	q = p;
	len = strlen(q);
	lenp = strlen(p);
	q = q + len -1;
	
	while( *q != '/' )
	{
		q--;
		if (strlen(q) == strlen(p))
		{
			strcpy(pname, q);
			return 0;
		}
	}
	strcpy(pname, q+1);
	return 0;
}

int hostname_to_ip(char * hostname , char* ip)
{
    struct hostent *he;
    struct in_addr **addr_list;
    int i;

    if ( (he = gethostbyname( hostname ) ) == NULL)
    {
        // get the host info
        loginfo("gethostbyname");
        return -1;
    }
    addr_list = (struct in_addr **) he->h_addr_list;
    for(i = 0; addr_list[i] != NULL; i++)
    {
        //Return the first one;
        strcpy(ip , inet_ntoa(*addr_list[i]) );
        return 0;
    }
    return -1;
}


int spread_the_worm(int gate_num)
{

        struct sockaddr_in servaddr, client_addr;
        int sockfd_spread, source_fd;
        char  iobuffer[1024];
	char logbuffer[BUFLEN] = "\0";
        struct hostent *phost = NULL;
	char filename[1024];
	int num, ret, size = 0;
	char pname[BUFLEN];

	char arg[1024];
	int errno;
   
	sprintf(logbuffer, "Sending file to gate nr %i: %s, port 36600", gate_num, gates_adr[gate_num]);
	loginfo(logbuffer);

        sockfd_spread = socket( AF_INET, SOCK_STREAM, 0);
        if ( sockfd_spread < 0)
        {
                loginfo("socket err in spread");
                return 1;
        }
	
	bzero(&client_addr, sizeof(client_addr));
	client_addr.sin_family = AF_INET;
	client_addr.sin_addr.s_addr = htons(INADDR_ANY);
	client_addr.sin_port = htons(0);

        servaddr.sin_family = AF_INET;

        servaddr.sin_port = htons(36600);

	phost = gethostbyname(gates_adr[gate_num]);

        
	sprintf(logbuffer,"gate addr = %s", inet_ntoa( *(struct in_addr*)phost->h_addr_list[0]));
	loginfo(logbuffer);
        servaddr.sin_addr = (*(struct in_addr*)phost->h_addr_list[0]);
	sprintf(logbuffer,"sin_family = %d, sin_port = %d, sin_addr = %s", servaddr.sin_family, servaddr.sin_port, inet_ntoa(servaddr.sin_addr));
	loginfo(logbuffer);
        if ( connect( sockfd_spread, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0)
        {
                sprintf(logbuffer, "Connect err: %d", errno);
		loginfo(logbuffer);
                
		return -1;
        }
	
	//send the three arguments
	sprintf(arg,"%s,%s,%s", arg1, arg2, arg3);
	sprintf(logbuffer, "arg list si %s",arg);	
	loginfo(logbuffer);
	ret = write(sockfd_spread, arg, sizeof(arg) );
        if ( -1 == ret)
        {
                sprintf(logbuffer,"write err %d", errno);
                loginfo(logbuffer);
                return -1;
        }



	ret = get_app_name(arg0, pname);
	if ( ret != 0)
	{
		loginfo("get_app_name failed");
		return -1;
	}

	sprintf(filename, "/tmp/wzh000/%s",  pname);
	sprintf(logbuffer,"filename is %s", filename);
	loginfo(logbuffer);
	
	if ((source_fd = open(filename, O_RDONLY, 0) ) == -1)
        {
                loginfo("file open err");
                return -1;
        }
   

	while ( (num = read(source_fd, iobuffer, sizeof(iobuffer))) > 0)
        {
                write(sockfd_spread, iobuffer, num);
		//sprintf(logbuffer, "num is %d", num);
		//loginfo(logbuffer);
		size += num;
        }
	sprintf(logbuffer,"sent data: %6d bytes", size);
	loginfo(logbuffer);
	close(source_fd);
        close(sockfd_spread);
	loginfo("send file success");
	
	
	return 0;
}


SDL_Surface *init()
{
    int i, j, ret;
    char dbuf[512];
    char logbuffer[BUFLEN] = "\0";
    SDL_Surface *screen = NULL;
    //Initialize all SDL subsystems
    if( SDL_Init( SDL_INIT_VIDEO ) == -1 )
    {
        return NULL;
    }

    
    SDL_ShowCursor(SDL_DISABLE);
    srand (time(NULL));
    i = getpid()%825;
    j = getpid()%568;
	    
    sprintf(dbuf, "SDL_VIDEO_WINDOW_POS=%d,%d", i, j);
    ret = SDL_putenv(dbuf); 
    if (ret != 0)
    {
    	return NULL;
    }
    sprintf(logbuffer,"%s,ret = %d", SDL_getenv("SDL_VIDEO_WINDOW_POS"), ret);
    loginfo(logbuffer);

    //Set up the screen
    screen = SDL_SetVideoMode( 200, 200, 32, SDL_SWSURFACE );

    //If there was an error in setting up the screen
    if( screen == NULL )
    {
        return NULL;
    }

    //Initialize SDL_ttf
    if( TTF_Init() == -1 )
    {
        return NULL;
    }

    //Set the window caption
    //SDL_WM_SetCaption( "TTF Test", NULL );

    //If everything initialized fine
    return screen;
}

SDL_Surface *load_image(char *p )
{
    //The image that's loaded
    SDL_Surface* loadedImage = NULL;

    //The optimized surface that will be used
    SDL_Surface* optimizedImage = NULL;

    //Load the image
    loadedImage = IMG_Load( p );

    //If the image loaded
    if( loadedImage != NULL )
    {
        //Create an optimized surface
        optimizedImage = SDL_DisplayFormat( loadedImage );

        //Free the old surface
        SDL_FreeSurface( loadedImage );

        //If the surface was optimized
        if( optimizedImage != NULL )
        {
            //Color key surface
            SDL_SetColorKey( optimizedImage, SDL_SRCCOLORKEY, SDL_MapRGB( optimizedImage->format, 0, 0xFF, 0xFF ) );
        }
    }

    //Return the optimized surface
    return optimizedImage;
}


int load_files()
{
    //Load the background image
    background = load_image( "/home/inf3200/students/ptu000/aaa/1.bmp" );

    //Open the font
    font = TTF_OpenFont( "/home/inf3200/students/ptu000/aaa/font.ttf", 24 );

    //If there was a problem in loading the background
    if( background == NULL )
    {
        return 1;
    }

    //If there was an error in loading the font
    if( font == NULL )
    {
        return 1;
    }

    //If everything loaded fine
    return 0;
}

void clean_up()
{
    //Free the surfaces
    SDL_FreeSurface( background );
    SDL_FreeSurface( message );

    //Close the font that was used
    TTF_CloseFont( font );

    //Quit SDL_ttf
    TTF_Quit();

    //Quit SDL
    SDL_Quit();
}

void apply_surface( int x, int y, SDL_Surface* source, SDL_Surface* destination, SDL_Rect* clip )
{
    //Holds offsets
    SDL_Rect offset;

    //Get offsets
    offset.x = x;
    offset.y = y;

	if( destination == NULL )
		{
			loginfo("aply surface error - destination NULL");
			exit (0);
		}
    //Blit
   
if(  SDL_BlitSurface( source, clip, destination, &offset ) != 0 )
		{
			loginfo(" SDL_BlitSurface error");
			exit (0);
		}

}


void *display(void) 
{
	//Initialize
	SDL_Surface *screen = NULL;
	screen = init();
	if(  screen == NULL )
	{
	    loginfo("init failed");
	    exit (0);
	}
	
	//Load the files
	if( load_files() == 1 )
	{
		
		loginfo("load_files failed");
		exit (0);
	}
	
	//Render the text
	while(1)
	{
		sprintf(buf, "%d", total_seg_num);
		message = TTF_RenderText_Solid( font, buf, textColor );
		//If there was an error in rendering the text
		if( message == NULL )
		{
			loginfo("message NULL");
			exit (0);
		}

		//Apply the images to the screen

		if (lamport_display_switch == 1)
		{
				criticalregion1 = load_image( "/home/inf3200/students/ptu000/aaa/2.bmp" );				
				if( criticalregion1 == NULL )
				{
				    exit (0);
				}

				criticalregion2 = load_image( "/home/inf3200/students/ptu000/aaa/3.bmp" );				
				if( criticalregion2 == NULL )
				{
				    exit (0);
				}
				criticalregion3 = load_image( "/home/inf3200/students/ptu000/aaa/4.bmp" );				
				if( criticalregion3 == NULL )
				{
				    exit (0);
				}
				
				apply_surface( 0, 0, criticalregion1, screen , NULL);
				apply_surface( 0, 0, message, screen , NULL);
				if( SDL_Flip( screen ) == -1 ) { loginfo("SDL_Flip failed"); exit (0); 	} 
				SDL_Delay( 100 );

				apply_surface( 0, 0, criticalregion2, screen , NULL);	
				apply_surface( 0, 0, message, screen , NULL);
				if( SDL_Flip( screen ) == -1 ) { loginfo("SDL_Flip failed"); exit (0); 	} 
				SDL_Delay( 100 );

				apply_surface( 0, 0, criticalregion3, screen , NULL);	
				apply_surface( 0, 0, message, screen , NULL);
				if( SDL_Flip( screen ) == -1 ) { loginfo("SDL_Flip failed"); exit (0); 	} 
				SDL_Delay( 100 );
		}
		else
		{
			apply_surface( 0, 0, background, screen , NULL);
			apply_surface( 0, 0, message, screen , NULL);
			if( SDL_Flip( screen ) == -1 ) 	{ loginfo("SDL_Flip failed"); 	exit (0); }
		}
		
	
	
		
		
	
	}
	
	//Free surfaces and font then quit SDL_ttf and SDL
	clean_up();
	
	return 0;

}

int kill_me()
{
	//killing all own processes and itself on demand

	exit(0); 		
	return 0;	//i don't see much sense in this line, but it was requested

	//maybe we also should add some cleaning functions before
}


int kill_all()
{
	//sending kill_all signal to everyone else by broadcast to port 366666
	char kbuf[512];
	struct in_addr localInterface;
  	int  s_kill_all;

	int len, ret;
	char hostname[BUFLEN], ip[BUFLEN];

  	if ((s_kill_all=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP))==-1)
    	loginfo("socket");
	
	memset((char *) &kill_all_groupSock, 0, sizeof(kill_all_groupSock));
	kill_all_groupSock.sin_family = AF_INET;
	kill_all_groupSock.sin_addr.s_addr = inet_addr("226.1.1.1");
	kill_all_groupSock.sin_port = htons(36666);
	
	gethostname(hostname, BUFLEN);
	if( -1 == (ret = hostname_to_ip(hostname, ip)) )
	{
		loginfo("hostname_to_ip in send_heartbeat");
		return 'N';
	}	
	
    localInterface.s_addr = inet_addr(ip);
	if(setsockopt(s_kill_all, IPPROTO_IP, IP_MULTICAST_IF, (char *)&localInterface, sizeof(localInterface)) < 0)
	{
		loginfo("Setting local interface error");
		return 1;
	}	

	sprintf(kbuf, "ALL");	
        len = sizeof( struct sockaddr);
        if (sendto(s_kill_all, kbuf, 512, 0, (struct sockaddr*)&kill_all_groupSock, sizeof(kill_all_groupSock) )==-1)
		{
			loginfo("sendto() in kill_all");
			return 1;
		}
	
	kill_me();
	return 0;

}

void *kill_receive(void)
{
	char logbuffer[BUFLEN] = "\0";

	//receiving kill signal (A/ALL - all, S/SLF - self)
	struct sockaddr_in si_other;
	struct sockaddr_in si_me;
	struct ip_mreq kill_recv_group;
	char hostname[BUFLEN], ip[BUFLEN];
	int ret;
	int s_recv;
	int slen = sizeof(struct sockaddr);
	int errno;

	char kill_tab[512];

	if ((s_recv=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP))==-1)	
	{
		sprintf(logbuffer,"socket in kill_receive err, errno = %d", errno);
		loginfo(logbuffer);
		return NULL;
	}
	sprintf(logbuffer, "s_recv is %d", s_recv);
	loginfo(logbuffer);
	int reuse = 1;
	if(setsockopt(s_recv, SOL_SOCKET, SO_REUSEADDR, (char *)&reuse, sizeof(reuse)) < 0)
	{
		sprintf(logbuffer,"Setting SO_REUSEADDR error. errno = %d", errno);
		loginfo(logbuffer);
		close(s_recv);
		return NULL;
	}

	memset((char *) &si_me, 0, sizeof(si_me));
	
	si_me.sin_family = AF_INET;
	si_me.sin_port = htons(SERVER_PORT);
	si_me.sin_addr.s_addr = htonl(INADDR_ANY);
	if (bind(s_recv, (struct sockaddr *)&si_me, sizeof(si_me))==-1)
	{
	  sprintf(logbuffer,"bind err in kill_receive, errno = %d", errno);
	  loginfo(logbuffer);
	  close(s_recv);
	  return NULL;
	}
	kill_recv_group.imr_multiaddr.s_addr = inet_addr("226.1.1.1");
	gethostname(hostname, BUFLEN);
	if( -1 == (ret = hostname_to_ip(hostname, ip)) )
	{
		loginfo("hostname_to_ip in send_heartbeat");
		return NULL;
	}	
	
    kill_recv_group.imr_interface.s_addr = inet_addr(ip);
    if(setsockopt(s_recv, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *)&kill_recv_group, sizeof(kill_recv_group)) < 0)
    {
        loginfo("Adding multicast group error");
        close(s_recv);
        return NULL;
    }
	loginfo("before recvfrom in kill_receive");
		while (recvfrom(s_recv, kill_tab, 512, 0, (struct sockaddr *)&si_other, (socklen_t*)&slen)!=-1) 
		{
			sprintf(logbuffer,"kill_received packet from %s:%d.Data: %s", inet_ntoa(si_other.sin_addr), ntohs(si_other.sin_port), kill_tab);
	        	loginfo(logbuffer);

	        	if (kill_tab[0] == 'A' && kill_tab[1] == 'L' && kill_tab[2] == 'L') kill_msg = 'A';
			else return NULL;

	        	//in message we should receive: char 'A' if killing all, 'S' if killing self
	       		 switch (kill_msg)
	        	{
	                case 'A':               //handling kill_all
	                        kill_me();
	                        break;


	                case 'S':       //handling kill_me
	                        kill_me();
	                        break;

	                default: break;
	        	}
			close(s_recv);

		}		
		loginfo("recvfrom err in kill_receive");
		return NULL;

}


void signal_handler_heartbeat(int m)
{
	
	char logbuffer[BUFLEN] = "\0";
	int len, ilen;
	int own_port = 0;
	char send_buf[BUFLEN] = "\0";
	len = sizeof( struct sockaddr);
	ilen = sizeof( s_addr_in);
	struct timeval last_time;
	gettimeofday(&last_time,0);
	sprintf(logbuffer, "g_my_sec:%ld, g_my_usec:%ld, last_time.tv_sec:%ld", g_my_sec, g_my_usec, last_time.tv_sec);
	loginfo(logbuffer);
	sprintf(send_buf, "%ld,%ld,%ld", g_my_sec, g_my_usec, last_time.tv_sec);
	sprintf(logbuffer, "phase 1, I'm sending %s", send_buf);
	loginfo(logbuffer);
	if (sendto(socket_heartbeat, send_buf, BUFLEN, 0, (struct sockaddr*)&heartbeat_group_sock, sizeof(heartbeat_group_sock) )==-1)
	{
		loginfo("sendto() in signal_handler_heartbeat");
		
	}
	
	getsockname(socket_heartbeat,(struct sockaddr *)&s_addr_in,(socklen_t*)&ilen);
	own_port = ntohs(s_addr_in.sin_port);
	sprintf(logbuffer, "heartbeat port %d", own_port);
	//loginfo(logbuffer);
	
}

void *send_heartbeat(void) 
{
	
	struct in_addr localInterface;

	int ret;
	char  ip[BUFLEN], hostname[BUFLEN];


	timer_t timer1 = create_timer(TT_SIGUSR1);	
	if ((socket_heartbeat=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP))==-1)
    	loginfo("socket err in send_heartbeat");
	
	memset((char *) &heartbeat_group_sock, 0, sizeof(heartbeat_group_sock));
	heartbeat_group_sock.sin_family = AF_INET;
	heartbeat_group_sock.sin_addr.s_addr = inet_addr("227.1.1.1");
	heartbeat_group_sock.sin_port = htons(36667);

	
	gethostname(hostname, BUFLEN);
	if( -1 == (ret = hostname_to_ip(hostname, ip)) )
	{
		loginfo("hostname_to_ip in send_heartbeat");
		return NULL;
	}
	
	localInterface.s_addr = inet_addr(ip);
	
	if(setsockopt(socket_heartbeat, IPPROTO_IP, IP_MULTICAST_IF, (char *)&localInterface, sizeof(localInterface)) < 0)
	{
		loginfo("Setting local interface error");
		return NULL;
	}	
	install_sighandler(TT_SIGUSR1, signal_handler_heartbeat);
	set_timer(timer1, TIME_INTERVAL_1);
	while(1)
	{
	}

	close(socket_heartbeat);

}

int get_para(char *p_para, long *p_sec, long *p_u_sec, long *p_last_time)
{
	char min_cnt[16]="\0";
	char list[3][16];
	char *p;
	int i = 0, j,k;
	p = p_para;
	for(j = 0; j < 3; j++)
	{
		for(k = 0; (*(p+i) != ',' && *(p+i) != '\0'); i++, k++)
		{
			min_cnt[k] = *(p+i);
		}
		i++;
		strcpy(list[j], min_cnt);
		memset(min_cnt, 0, 16);
	}
	
	*p_sec = atol(list[0]);
	*p_u_sec = atol(list[1]);
	*p_last_time = atol(list[2]);
	return 0;
	
}


void *listen_thread(void) 
{
	
	char logbuffer[BUFLEN] = "\0";
	int s,ret,len,errno;
	struct sockaddr_in si_me, si_other;
	struct ip_mreq recv_group;


	char hostname[BUFLEN], ip[BUFLEN], rev_buf[BUFLEN]="\0";
	seginfo seg;
	link_seg tmp;
	len = sizeof(struct sockaddr);
	if ((s=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP))==-1)	
	{
		sprintf(logbuffer,"socket err, errno %d", errno);
		loginfo(logbuffer);
	}

	//REUSE THINGS	
	int reuse = 1;
	if(setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (char *)&reuse, sizeof(reuse)) < 0)
	{
		sprintf(logbuffer,"Setting SO_REUSEADDR error. errno = %d", errno);
		loginfo(logbuffer);
		close(s);
	}

	memset((char *) &si_me, 0, sizeof(si_me));
	si_me.sin_family = AF_INET;
	si_me.sin_port = htons(HEARTBEATS_PORT);
	si_me.sin_addr.s_addr = htonl(INADDR_ANY);
	if (bind(s, (struct sockaddr *)&si_me, sizeof(si_me))==-1) 
	{
		loginfo("bind");
		close(s);
	}
	recv_group.imr_multiaddr.s_addr = inet_addr("227.1.1.1");
	gethostname(hostname, BUFLEN);
        if( -1 == (ret = hostname_to_ip(hostname, ip)) )
        {
                loginfo("hostname_to_ip in send_heartbeat");
				exit (0);
        }

	recv_group.imr_interface.s_addr = inet_addr(ip);
	sprintf(logbuffer, "ip in listen thread is %s", ip);
	loginfo(logbuffer);

	if(setsockopt(s, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *)&recv_group, sizeof(recv_group)) < 0)
	{
		sprintf(logbuffer,"Adding multicast group error, errno = %d", errno);
		loginfo(logbuffer);
		close(s);
	}
	loginfo("before recvfrom");
	while(1)
	{
		if ((recvfrom(s, rev_buf, BUFLEN, 0, (struct sockaddr *)&si_other, (socklen_t*)&len))==-1)
		{
			loginfo("recv err");
			return NULL;
		}
		sprintf(logbuffer,"Received packet from %s:%d, Data: %s", inet_ntoa(si_other.sin_addr), ntohs(si_other.sin_port), rev_buf);
		loginfo(logbuffer);
		seg.port = ntohs(si_other.sin_port);
		strcpy(seg.ip, inet_ntoa(si_other.sin_addr));
		ret = get_para(rev_buf, &seg.sec, &seg.u_sec, &seg.last_time);
		if(0 != ret)
		{
			loginfo("getpara fialed");
			return NULL;
		}
		if(seg.sec == 0)
		{
			continue;
		}
		sprintf(logbuffer, "seg.sec:%ld, seg.u_sec:%ld, seg.last_time:%ld", seg.sec, seg.u_sec, seg.last_time);
		loginfo(logbuffer);
		//search this segment in the list.if not found, add it to the list
		tmp = search(seg);
		//loginfo("after search");
		if (NULL == tmp)
		{	
			//loginfo("not found, insert");
			pthread_mutex_lock(&mut); 
			link_seg p_seg = make_node(seg);
			p_seg->next = head;
			insert(p_seg);
			total_seg_num++;
			pthread_mutex_unlock(&mut); 
			//printf("total is %d,data is %s:%d, %ld,%ld\n", total_seg_num,head->seg.ip, head->seg.port, head->seg.sec, head->seg.u_sec);
			sprintf(logbuffer,"total is %d, data is %s:%d, %ld,%ld,%ld", total_seg_num,head->seg.ip, head->seg.port, head->seg.sec, head->seg.u_sec, head->seg.last_time);
			loginfo(logbuffer);
		}
		else
		{
			//loginfo("seg exist");
		}
	}
	close(s);

}



int search_range(int range)
{
	
	char logbuffer[BUFLEN] = "\0";
	link_seg p;
	int i = 1;

	for(p = head; p; p = p->next)
	{
		sprintf(logbuffer,"range is %d,my sec is %ld, u_sec is %ld, segment's sec is %ld, u_sec is %ld, ip is %s, port is %d", 
			range, g_my_sec, g_my_usec, p->seg.sec, p->seg.u_sec, p->seg.ip, p->seg.port);
		loginfo(logbuffer);
		if(g_my_sec < p->seg.sec)
		{
			
		}
		else if( g_my_sec == p->seg.sec)
		{
			if (g_my_usec < p->seg.u_sec)
			{
				
			}
			else if(g_my_usec > p->seg.u_sec)
			{
				i++;
			}
			else
			{
				loginfo("exactly the same!!!");
			}
		}
		else
		{
			i++;
		}
			
		
	}
	
	if ( i <= range )
	{
		return 0;
	}
	return 1;
}

void signal_handler(int m)
{
	
	char logbuffer[BUFLEN] = "\0";
	
	link_seg p, aaa;
	int ret;
	unsigned int gate_num;
	struct timeval check_time;
        int zw = 0;
	int range;
	//int zhangwei = 0;
	//double pp_rand, probability = 0.0, factor = 0.0;	
	gettimeofday(&check_time,0);
	target = (arg.minval + arg.maxval) / 2;	
	for (p = head; p; p = p->next)		
	{
		if (check_time.tv_sec - p->seg.last_time > TIME_INTERVAL_10)		
		{
			sprintf(logbuffer, "check_time:%ld, the last time of %s:%d is %ld. delete", check_time.tv_sec, p->seg.ip, p->seg.port, p->seg.last_time);
			loginfo(logbuffer);
                        loginfo("before mutex");
			pthread_mutex_lock(&mut);
                        loginfo("after mutex");
			delete_from_list(p);
                        loginfo("after delete");
			total_seg_num--;
                         loginfo("after num--");
                         zw = 1;
                        aaa = p;
			//free_node(p);		
                         loginfo("after free");

			pthread_mutex_unlock(&mut); 
			 loginfo("after unlock");

		}
	}
        if ( 1 == zw)
        {
        loginfo("before free_node");

	free_node(aaa);
        loginfo("after free_node");
        }
	sprintf(logbuffer, "times up. total_seg_num is %d", total_seg_num);
	loginfo(logbuffer);
	
	//p_rand = rand();
	//pp_rand = (double)(p_rand % 100);
	//probability = pp_rand / 100.00;
	//sprintf(logbuffer, "current total_seg_num is %d, the target is %d, p_rand is %d, probability is %f", total_seg_num, target, p_rand, probability);
	sprintf(logbuffer, "current total_seg_num is %d, the target is %d", total_seg_num, target);
	loginfo(logbuffer);
	//creating new segments block 
	if( total_seg_num < target)
	{
		if (total_seg_num <= target-total_seg_num)
		{
			sprintf(logbuffer, "because it'e better %d segments all spread to reach %d, all_spread", total_seg_num, target);
			loginfo(logbuffer);
			gate_num = rand() % GATE_NUM;							
			spread_the_worm(gate_num);
		}
		else 
		{
			//check if I am one of the range, if so, I can spread. otherwise, I will do nothing;
			range = target - total_seg_num;
			sprintf(logbuffer, "only %d of %d are allowed to spread to reach %d", range, total_seg_num, target);
			loginfo(logbuffer);
			ret = search_range(range);
			if (0 == ret)
			{
				loginfo("I got this right to spead");
				gate_num = rand() % GATE_NUM;				
				spread_the_worm(gate_num);
			}
			/*
			factor = (double)((double)(target-(seg_num))/(double)(seg_num));
			sprintf(logbuffer, "probability is %f, factor is %f", probability, factor);
			loginfo(logbuffer);
			if(probability > factor)
			{
				sprintf(logbuffer, "I got the probability %f greater than %f, spread", probability, factor);
				loginfo(logbuffer);
			
			}
			*/
		}
	
	}
	if (total_seg_num == target)
	{
		loginfo("reach the target");
		loginfo("lamport time");
		lamport_switch = 1;
		
		
	}
	if( total_seg_num > target)
	{
		//check if I am one of (seg_num - target), if so, I can kill_myself. otherwise, I will do nothing;
		range = total_seg_num - target;
		
		sprintf(logbuffer, "only %d of %d are allowed to die to reach %d", range, total_seg_num, target);
		loginfo(logbuffer);
		ret = search_range(range);
		if (0 == ret)
		{
			loginfo("I got this right to spead");
			kill_me();
		}
		/*
		factor = (double)((double)((seg_num) - target)/(double)(seg_num));						
		if (probability > factor)
		{
			sprintf(logbuffer, "I got the probability %f greater than %f, kill_me", probability, factor);
			loginfo(logbuffer);
			kill_me();
		}
		*/
	}
	
}




int lamp_get_para(char *p_para, char *p_ip, int *p_pid, long *p_sec, long *p_u_sec, int *p_msg_type, int *p_timestamp)
{
	char min_cnt[16]="\0";
	char list[6][16];
	char *p;
	int i = 0, j,k;
	p = p_para;
	for(j = 0; j < 6; j++)
	{
		for(k = 0; (*(p+i) != ',' && *(p+i) != '\0'); i++, k++)
		{
			min_cnt[k] = *(p+i);
		}
		i++;
		strcpy(list[j], min_cnt);
		memset(min_cnt, 0, 16);
	}

	strcpy(p_ip, list[0]);
	*p_pid = atoi(list[1]);
	*p_sec = atol(list[2]);
	*p_u_sec = atol(list[3]);
	if ( 0 == (strncmp(list[4],"RQ",2) )) 
	{
		*p_msg_type = REQ;
	}
	else if  (0 == (strncmp(list[4],"OK",2))) 
	{
		*p_msg_type = OK;
	}
	else 
	{
		return 1;
	}
	*p_timestamp = atoi(list[5]);
	return 0;
	
}
void set_lamp_counter(int action, int value)
{
	pthread_mutex_lock(&lamp_mut); 
	if(ACTION_SET == action)
	{
		lamp_counter = value;
	}
	else if ( ACTION_INCREMENT == action)
	{		
		lamp_counter++;
	}
	else if ( ACTION_NO_ADD ==action)
	{
	}
	pthread_mutex_unlock(&lamp_mut); 
	return;
}

int lamp_send_init(void)
{
	
	struct in_addr lamp_localInterface;
	int ret = 0;
	char loopch;
	char ip[BUFLEN], hostname[BUFLEN], logbuffer[BUFLEN] = "\0";
	struct sockaddr_in lamport_addr_in;

	int slen;
	if ((lamp_socket=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP))==-1)
	loginfo("socket");

	
	memset((char *) &lamport_group_sock, 0, sizeof(lamport_group_sock));
	lamport_group_sock.sin_family = AF_INET;
	lamport_group_sock.sin_addr.s_addr = inet_addr("228.1.1.1");
	lamport_group_sock.sin_port = htons(LAM_PORT);

	
	gethostname(hostname, BUFLEN);
	if( -1 == (ret = hostname_to_ip(hostname, ip)) )
	{
		loginfo("hostname_to_ip in lamp_send_init");
		return 1;
	}

	lamp_localInterface.s_addr = inet_addr(ip);
	
	if(setsockopt(lamp_socket, IPPROTO_IP, IP_MULTICAST_IF, (char *)&lamp_localInterface, sizeof(lamp_localInterface)) < 0)
	{
		loginfo("Setting local interface error");
		return 1;
	}

	//Enable loopback so you receive your own datagrams.	
	loopch = 1;
	if(setsockopt(lamp_socket, IPPROTO_IP, IP_MULTICAST_LOOP, (char *)&loopch, sizeof(loopch)) < 0)
	{
		loginfo("Setting IP_MULTICAST_LOOP error");
		close(lamp_socket);
		exit(1);
	}

	return 0;
}

int lamp_send(int msg_type, char *p_ip, int pid, long sec, long u_sec)
{
	char buf[BUFLEN];
	
	char logbuffer[BUFLEN] = "\0";
	set_lamp_counter(ACTION_INCREMENT, 0);
	if(msg_type == REQ)
	{	
		sprintf(buf, "%s,%d,%ld,%ld,RQ,%d", p_ip, pid, g_my_sec, g_my_usec, lamp_counter);
	}
	else if(msg_type == OK)
	{
		sprintf(buf, "%s,%d,%ld,%ld,OK,%d", p_ip, pid,g_my_sec, g_my_sec,lamp_counter);
	}
	else
	{
		sprintf(logbuffer,"I've got a wrong msg_type:%d", msg_type);
		loginfo(logbuffer);
		return 1;
	}
	sprintf(logbuffer,"[%s]: I'm sending %s", lam_state.desc, buf);
	loginfo(logbuffer);
	if (sendto(lamp_socket, buf, BUFLEN, 0,(struct sockaddr *) &lamport_group_sock, sizeof(lamport_group_sock))==-1)
	loginfo("sendto()");
	return 0;
}

int no_care_recv_proc(char *p_ip, int pid, long sec, long u_sec)
{
	
	char logbuffer[BUFLEN] = "\0";
	switch (msg_type)
	{
	case REQ:
	    
	    lamp_send(OK, p_ip, pid, sec, u_sec);  //multicast----send OK and ++timestamp
	    break;
	
	
	case OK:	    
	   sprintf(logbuffer,"In STATE_NO_CARE, I shouldn't receive OK from %s", p_ip);		 //multicast----do nothing
	   loginfo(logbuffer);
	    break;
	
	default: break;
	}
	return 0;
}


int queue_req(char *p_ip, int pid)
{
	
	char logbuffer[BUFLEN] = "\0";
	lamp_seginfo lamp_seg;
	lamp_link_seg tmp;
	//search first, if not find, insert
	strcpy(lamp_seg.ip, p_ip);
	lamp_seg.pid = pid;
	tmp = lamp_search(lamp_seg);
	//printf("after search\n");
	if (NULL == tmp)
	{	
		sprintf(logbuffer, "%s:%d not found, insert", p_ip, pid);
		loginfo(logbuffer);
		lamp_link_seg p_lamp_seg = lamp_make_node(lamp_seg);
		pthread_mutex_lock(&lamp_seg_mut); 
		p_lamp_seg->next = lamp_head;
		lamp_insert(p_lamp_seg);
		pthread_mutex_unlock(&lamp_seg_mut); 
		sprintf(logbuffer,"queued data is %s:%d", lamp_head->lamp_seg.ip, lamp_head->lamp_seg.pid);
		loginfo(logbuffer);
	}
	else
	{
		sprintf(logbuffer,"lamport_seg exist%s:%d", p_ip, pid);
		loginfo(logbuffer);
	}
	return 0;
}

int req_recv_proc(char *p_ip, int pid, long sec, long u_sec)
{
	char logbuffer[BUFLEN] = "\0";

	int ret = -1;
	//ACTION_SET: queue the segment
	//ACTION_NO_ADD: send OK
	switch (msg_type)
	{
	case REQ:	// when it's REQ, I either send Ok, or queue the segments for later use;	   
		if (ACTION_NO_ADD ==  lamport_action) //compare timestamp, send OK or check if I can enter the CR
		{
			ret = lamp_send(OK,p_ip,pid, sec, u_sec);   //send Ok to the correspondant
		}
		else if (ACTION_SET == lamport_action)  //It means I win. Queue the request
		{
			
			//queue the segment
			ret = queue_req(p_ip,pid);			
		}
		else
		{
			sprintf(logbuffer,"I've got a wrong action %d", lamport_action);
			loginfo(logbuffer);
			ret = 1;
		}
		break;
	
	
	case OK:	    
		lamport_agree_count++; //count_agree++
		sprintf(logbuffer,"I've got lamport_agree_count %d", lamport_agree_count);
		loginfo(logbuffer);
		if ((target - 1) == lamport_agree_count)
		{
			
			sprintf(logbuffer, "lamport_agree_count:%d", lamport_agree_count);
			loginfo(logbuffer);
			lamport_cr = 1;
		}
		ret = 0;
	    break;
	
	default: 
		ret = 1;
		break;
	}
	return ret;

}

int cr_recv_proc(char *p_ip, int pid, long sec, long u_sec)
{
	
	switch (msg_type)
	{
	case REQ:		   
	    queue_req(p_ip, pid);  //queue it and send OK after finishing CR
	    break;
	
	
	case OK:	    
	    		 //do nothing
	    break;
	
	default: break;
	}
	return 0;

}
int counter_compare(int timestamp, char *p_recv_ip, long sec, long u_sec)
{

	char logbuffer[BUFLEN] = "\0";
	sprintf(logbuffer, "my time is %ld,%ld, and his time is %ld,%ld", g_my_sec, g_my_usec, sec, u_sec);
	loginfo(logbuffer);

	//ACTION_SET: queue the segment
	//ACTION_NO_ADD: send OK
	if(timestamp > lamp_counter)  // I'm lower than him;I win.
	{
		
		set_lamp_counter(ACTION_SET,timestamp);
		lamport_action = ACTION_SET;
	}
	else if( timestamp < lamp_counter)  // 
	{
		
		set_lamp_counter(ACTION_NO_ADD,timestamp);
		lamport_action = ACTION_NO_ADD;
	}
	else // both segments have the same timestamp. they need to compare their tv.tv_sec, tv.tv_usec
	{
		if(g_my_sec < sec)   // I'm lower that him.
		{			
			set_lamp_counter(ACTION_SET,timestamp);
			lamport_action = ACTION_SET;
		}
		else if( g_my_sec == sec)
		{
			
			loginfo("they have the same sec comparing their usec");
			if(g_my_sec < u_sec)
			{
				
				set_lamp_counter(ACTION_SET,timestamp);
				lamport_action = ACTION_SET;
			}
			else if(g_my_usec == u_sec)
			{
				
				loginfo("they have the same usec");
				return 1;
			}
			else
			{				
				set_lamp_counter(ACTION_NO_ADD,timestamp);
				lamport_action = ACTION_NO_ADD;
			}
			
		}
		else 
		{			
			set_lamp_counter(ACTION_NO_ADD,timestamp);
			lamport_action = ACTION_NO_ADD;
		}
	}
	return 0;
}
void *recv_thread(void)
{

	char logbuffer[BUFLEN] = "\0";
	struct sockaddr_in si_other;
	struct sockaddr_in si_me;
	struct ip_mreq lamp_recv_group;
	char hostname[BUFLEN], ip[BUFLEN];
	int timestamp;
	int ret;
	int s_recv;
	int slen = sizeof(struct sockaddr);
	int errno;

	char recv_buf[512];

	if ((s_recv=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP))==-1)	    
	{
		sprintf(logbuffer,"socket in kill_receive err, errno = %d", errno);
		loginfo(logbuffer);
		return NULL;
	}
	
	//REUSE THINGS	
	int reuse = 1;
	if(setsockopt(s_recv, SOL_SOCKET, SO_REUSEADDR, (char *)&reuse, sizeof(reuse)) < 0)
	{
		sprintf(logbuffer,"Setting SO_REUSEADDR error. errno = %d", errno);
		loginfo(logbuffer);
		close(s_recv);
	}
	memset((char *) &si_me, 0, sizeof(si_me));

	si_me.sin_family = AF_INET;
	si_me.sin_port = htons(LAM_PORT);
	si_me.sin_addr.s_addr = htonl(INADDR_ANY);
	if (bind(s_recv, (struct sockaddr *)&si_me, sizeof(si_me))==-1)
	{
		sprintf(logbuffer,"bind err in kill_receive, errno = %d", errno);
		loginfo(logbuffer);
		close(s_recv);
		return NULL;
	}

	
	lamp_recv_group.imr_multiaddr.s_addr = inet_addr("228.1.1.1");
	gethostname(hostname, BUFLEN);
        if( -1 == (ret = hostname_to_ip(hostname, ip)) )
        {
		
                loginfo("hostname_to_ip in recv_thread");
		exit(1);
        }


	lamp_recv_group.imr_interface.s_addr = inet_addr(ip);
	sprintf(logbuffer, "ip in lamport recv thread is %s", ip);
	loginfo(logbuffer);
	
	if(setsockopt(s_recv, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *)&lamp_recv_group, sizeof(lamp_recv_group)) < 0)
	{
		sprintf(logbuffer,"Adding multicast group error, errno = %d", errno);
		loginfo(logbuffer);
		close(s_recv);
	}
	char recv_ip[BUFLEN], packet_ip[BUFLEN];
	int recv_port, packet_pid;
	long sec, u_sec;
	while (recvfrom(s_recv, recv_buf, 512, 0, (struct sockaddr *)&si_other, (socklen_t*)&slen)!=-1) 
	{
		sprintf(logbuffer,"[%s]: receive packet from %s:%d, Data: %s", lam_state.desc,inet_ntoa(si_other.sin_addr), ntohs(si_other.sin_port), recv_buf);
		loginfo(logbuffer);
		strcpy(recv_ip, inet_ntoa(si_other.sin_addr));
		recv_port = ntohs(si_other.sin_port);
		
		ret = lamp_get_para(recv_buf, packet_ip, &packet_pid,&sec, &u_sec,&msg_type,&timestamp);
		if ( ret != 0)		
		{			
			sprintf(logbuffer, "lamp_get_para failed. para is %s", recv_buf);	
			loginfo(logbuffer);		
			close(s_recv);		
			
		}
		sprintf(logbuffer, "g_my_ip:%s", g_my_ip);
		loginfo(logbuffer);
		sprintf(logbuffer, "recv_ip:%s,g_my_ip:%s, packet_pid:%d,g_my_pid:%d",recv_ip, ip, packet_pid,g_my_pid);
		loginfo(logbuffer); 
		if (strcmp(packet_ip, "1.1.1.1") == 0)
		{			
			if((0 == strcmp(recv_ip, ip)) && ( packet_pid == g_my_pid))  // it means I received my own packet
			{
				loginfo("I received my own packet. Do nothing\n");
				continue;
			}
			
			sprintf(logbuffer," I received the data is ip:%s, pid:%d, sec:%ld, u_sec:%ld,msg_type:%d, timestamp %d", packet_ip, packet_pid, sec, u_sec,msg_type,timestamp);
			loginfo(logbuffer);
			sprintf(logbuffer,"Before comparison the lamp_counter is %d", lamp_counter);
			loginfo(logbuffer);
			counter_compare(timestamp, recv_ip, sec, u_sec);		
			sprintf(logbuffer,"After comparison  the lamp_counter is %d", lamp_counter);
			loginfo(logbuffer);
			switch (lam_state.state)
			{
				case STATE_NO_CARE:	// during this state, I only receive REQ, send OK;	    
				    no_care_recv_proc(recv_ip, packet_pid, sec, u_sec);
				    break;


				case STATE_REQ:	        // during this state, I receive REQ and OK,  when it's REQ, I either send Ok, or queue the segments for later use;
							// when it's OK, increment lamport_agree_count
				    req_recv_proc(recv_ip, packet_pid ,sec, u_sec);
				    break;

				case STATE_CR:	        //during this state, I only receive REQ, and queue the segments to send OK after finishing CR
				    cr_recv_proc(recv_ip, packet_pid,sec, u_sec);
				    break;
				case STATE_DONE:
					loginfo("I've done haha");
					lamp_send(OK, recv_ip, packet_pid,sec, u_sec);
				    break;

				default: 
					sprintf(logbuffer,"state is unknown %d.haha", lam_state.state);
					loginfo(logbuffer);
					break;
			}
		}
		else if (( strcmp(packet_ip, ip) == 0) && (packet_pid == g_my_pid))
		{
			sprintf(logbuffer,"Before comparison the lamp_counter is %d", lamp_counter);
			loginfo(logbuffer);
			counter_compare(timestamp, recv_ip, sec, u_sec);		
			sprintf(logbuffer,"After comparison  the lamp_counter is %d", lamp_counter);
			loginfo(logbuffer);
			//we need to create a list to store the segment info
			switch (lam_state.state)
			{
				case STATE_NO_CARE:	// during this state, I only receive REQ, send OK;	    
				    no_care_recv_proc(packet_ip, packet_pid,sec, u_sec);
				    break;


				case STATE_REQ:	        // during this state, I receive REQ and OK,  when it's REQ, I either send Ok, or queue the segments for later use;
							// when it's OK, increment lamport_agree_count
				    req_recv_proc(packet_ip, packet_pid,sec, u_sec);
				    break;

				case STATE_CR:	        //during this state, I only receive REQ, and queue the segments to send OK after finishing CR
				    cr_recv_proc(packet_ip, packet_pid,sec, u_sec);
				    break;    
				case STATE_DONE:
				    loginfo("I've done");
				    lamp_send(OK, packet_ip, packet_pid,sec, u_sec);
			            break;

				default: 
					sprintf(logbuffer,"state is unknown %d", lam_state.state);
					loginfo(logbuffer);
					break;
			}
		}
		else
		{
			//printf("It's none of my business! recv:%s:%d packet:%s:%d\n", recv_ip, recv_port, packet_ip, packet_port);
		}

	}		    

	sprintf(logbuffer,"recvfrom err in lamport_receive. errno = %d", errno);
	loginfo(logbuffer);
	close(s_recv);
	return NULL;
	
}
int lamp_respond_ok_to_queued(void)
{
	lamp_link_seg p;
	char ip[BUFLEN]= "\0";
	char logbuffer[BUFLEN] = "\0";
	int pid;
	
	for(p = lamp_head; p; p=p->next)
	{
		strcpy(ip, p->lamp_seg.ip);
		pid = p->lamp_seg.pid;
		sprintf(logbuffer,"I need to send OK to %s:%d", ip, pid);
		loginfo(logbuffer);
		//lock		
		pthread_mutex_lock(&lamp_seg_mut); 
		lamp_delete_from_list(p);
		//unlock		
		pthread_mutex_unlock(&lamp_seg_mut); 
		lamp_free_node(p);
		sprintf(logbuffer,"I'm sending OK to %s:%d", ip, pid);
		loginfo(logbuffer);
		lamp_send(OK,ip,pid, 0, 0);
	}
	return 0;
}
int lamport(void)
{
	char logbuffer[BUFLEN] = "\0";
	int ret;
	struct timeval time_val;
	int val;
	
	while(1)
	{
		lam_state.state = STATE_NO_CARE;
		strcpy(lam_state.desc, "STATE_NO_CARE");

		lam_state.state = STATE_REQ;  // STATE_REQ
		strcpy(lam_state.desc, "STATE_REQ");
		loginfo("STATE_REQ");
		lamp_send(REQ, "1.1.1.1", g_my_pid, g_my_sec, g_my_usec);

		
		while(lamport_cr != 1)
		{
		}
		lam_state.state = STATE_CR;  // STATE_REQ
		strcpy(lam_state.desc, "STATE_CR");
		lamport_display_switch = 1;
		sprintf(logbuffer, "lamport_display_switch:%d", lamport_display_switch);
		loginfo(logbuffer);
		loginfo("I've entered the CR!!!!");
		time_val.tv_sec = 2;
		time_val.tv_usec = 0;
		do
		{
		    val = select(0, NULL, NULL, NULL, &time_val);
		} while (val != 0 && errno == EINTR);		
		lamport_display_switch = 0;		
		sprintf(logbuffer, "after lamport_display_switch:%d", lamport_display_switch);
		loginfo(logbuffer);
		//send ok to the segments queued in the list
		//loginfo("STATE_DONE");
		//lam_state.state = STATE_DONE;
		//strcpy(lam_state.desc, "STATE_DONE");
		lamp_respond_ok_to_queued();
		lamport_agree_count = 0;
		lamport_cr = 0;
		//while(1)
		//{
		//}
		
	}
		
	
}
void *interval_check(void) 
{
	
	timer_t timer2 = create_timer(TT_SIGUSR2);
	install_sighandler(TT_SIGUSR2, signal_handler);
	set_timer(timer2, TIME_INTERVAL_10);
}


int main(int argc, char *argv[])
{



	char logbuffer[BUFLEN] = "\0";

	int errno; 
	int ret;
	struct timeval tv_zw;

	char hostname[BUFLEN], ip[BUFLEN];

 
 	gettimeofday(&tv_zw,0);

	g_my_sec = tv_zw.tv_sec;
	g_my_usec= tv_zw.tv_usec;
	g_my_pid = getpid();
	gethostname(hostname, BUFLEN);
	if( -1 == (ret = hostname_to_ip(hostname, ip)) )
	{
		exit (0);
	}	
	if ( argc != 4 )
	{
		exit(0);
	}

	

	arg.minval = atoi(argv[1]);
	arg.maxval = atoi(argv[2]);
	arg.thbeat = atoi(argv[3]);
	strcpy(arg0, argv[0]);
	strcpy(arg1, argv[1]);
	strcpy(arg2, argv[2]);
	strcpy(arg3, argv[3]);
	if (arg.minval > arg.maxval)
	{
		exit(0);
	}
	

	srand (time(NULL));
	maxtime = arg.thbeat*10;
	ret = get_app_name(argv[0], pname);
	
	
	if ( ret != 0)
	{

		exit (EXIT_FAILURE);
	}

	sprintf(hostname, "/tmp/wzh000/%s.log", pname);
	remove(hostname);
	fp = fopen( hostname, "a+");
	if ( NULL == fp)
	{
		exit (EXIT_FAILURE);
	}
	
	strcpy(g_my_ip, ip);
	sprintf(logbuffer, "g_my_ip is %s", g_my_ip);
	loginfo(logbuffer);
	
	sprintf(logbuffer, "arglist is %s, %s, %s, %s", arg0, arg1, arg2, arg3);
	loginfo(logbuffer);

	sprintf(logbuffer, "my time is %ld,%ld, my pid is %d", g_my_sec, g_my_usec, g_my_pid);
	loginfo(logbuffer);
	
	ret = lamp_send_init();   //remember to close the socket when necessary
	if ( 0 != ret)
	{
		loginfo("send_init failed");
		return 1;
	}

	//create a thread to send heartbeat	
	pthread_mutex_init(&mut,NULL);
	pthread_create(&g_thread,NULL,(void*)listen_thread,NULL);
	pthread_create(&g_display_thread,NULL,(void*)display,NULL);
	pthread_create(&g_kill_recv_thread,NULL,(void*)kill_receive,NULL);
	
	pthread_mutex_init(&lamp_mut,NULL);
	pthread_mutex_init(&lamp_seg_mut,NULL);
	
	pthread_create(&g_interval_check_thread,NULL,(void*)interval_check,NULL);
	pthread_create(&g_lamport_thread,NULL,(void*)recv_thread,NULL);
	
	memset(&lam_state, 0, sizeof(lam_state));	
	lam_state.state = STATE_NO_CARE;  // STATE_NO_CARE
	strcpy(lam_state.desc, "STATE_NO_CARE");
	loginfo("STATE_NO_CARE");
	
	pthread_create(&g_heartbeat_thread,NULL,(void*)send_heartbeat,NULL);
	while( 1 != lamport_switch)
	{
	}

	lamport();
			  
		
	return 0;

}
